本文是字节跳动数据平台开发套件团队在Flink Forward Asia 2021: Flink Forward 峰会上的演讲，着重分享了字节跳动数据湖技术上的选型思考和探索实践。本文重点分享字节的探索实践，对话框回复数字9可以阅读关于技术选型的思考。
文 | Gary Li 字节跳动数据平台开发套件团队高级研发工程师，数据湖开源项目Apache Hudi PMC Member在选择了基于Hudi的数据湖框架后，我们基于字节跳动内部的场景，打造定制化落地方案。我们的目标是通过Hudi来支持所有带Update的数据链路：
目标明确后，我们开始了对Hudi Flink Writer进行了测试。这个图是Hudi on Flink Writer的架构：一条新的数据进来之后，首先会经过一个索引层，从而找到它需要去的地方。
- State索引中保存了所有主键和文件ID的一一映射关系，对于Update数据，会找到其所存在的文件ID，对于Insert数据，索引层会给他指定一个新的文件ID，或者是历史文件中的小文件，让其填充到小文件中，从而避免小文件问题。
- 经过索引层之后，每条数据都会带有一个文件ID，Flink会根据文件ID进行一次shuffle，将相同文件ID的数据导入到同一个子任务中，同时可以避免多个任务写入同一个文件的问题。
- 写入子任务中有一个内存缓冲区，用于储存当前批次的所有数据，当Checkpoint触发时，子任务缓冲区的数据会被传入Hudi Client中，Client会去执行一些微批模式的计算操作，比如Insert/Upsert/Insert overwrite等，每种操作的计算逻辑不同，比如说Insert操作，会生成一个新的文件，Upsert操作可能会和历史文件做一次合并，
- 待计算完成后，将处理好的数据写入到HDFS中，并同时收集元数据。
- Compaction任务为流任务的一部分，会定时的去轮训Hudi的时间线，查看是否有Compaction计划存在，如果有Compaction计划，会通过额外的Compaction算子来执行。
- 在数据量比较大的场景下，所有的主键和文件ID的映射关系都会存在State中，State的体积膨胀的非常快，带来了额外的储存开销，并且有时会造成Checkpoint超时的问题。
- 第二个问题是，由于Checkpoint期间，Hudi Client操作比较重，比如说和底层的base文件进行合并，这种操作涉及到了历史文件的读取，去重，以及写入新的文件，如果遇到HDFS的抖动，很容易出现Checkpoint超时的问题
- 第三个问题是，Compaction任务作为流式任务的一部分，任务启动后资源就不可调节，如果需要调节，只能重启整个任务，开销比较大，如果不能灵活调节Compaction任务，就可能会出现Compaction算子空跑导致资源浪费，或者资源不足导致任务失败的情况
为了解决这些问题，我们开始针对我们的场景进行了定制化的优化索引的目的就是找到当前这条数据所在的文件地点，存在State中的话每条数据都涉及到一次State的读和写，在数据量大的场景下，所带来的计算和储存开销都是比较大的
字节跳动内部开发了一种基于哈希的索引，可以通过直接对主键的哈希操作来找到文件所在的位置，这种方式在非分区表下可以做到全局索引，绕过了对State的依赖，改造过后，索引层变成了一层简单的哈希操作。早期的Hudi写入和Spark强绑定，在2020年底，Hudi社区对底层的Hudi Client进行了拆分，并且支持了Flink引擎，这种改造方式是将Spark RDD的操作变成了一个List的操作，所以底层还是一个批式操作，对于Flink来说，每一次Checkpoint期间所需要做的计算逻辑是类似于Spark RDD的，相当于是执行了一次批式的操作，计算包袱是比较大的。写入层的具体流程是：一条数据经过索引层后，来到了写入层，数据首先会在Flink的内存缓冲区积攒，同时通过内存监控来避免内存超出限制导致任务失败，到了Checkpoint的时候，数据会被导入到Hudi Client，然后Hudi Client会通过Insert，Append，Merge等操作计算最终的写入数据，计算完成后将新的文件写入到HDFS并同时回收元数据。我们的核心目标在于如何让这种微批的写入模式更加的流式化，从而降低Checkpoint期间的计算负担。- 在表结构上，我们选择了与流式写入更加匹配的Merge on Read格式，写入的算子只负责对于log文件的追加写入，不做任何别的额外的操作，例如和base文件进行合并。
- 在内存上，我们将第一层Flink的内存缓冲区去掉，直接把内存缓冲区建立在了hudi client中，在数据写入的同时进行内存监控避免内存超出限制的情况，
- 我们将写入hdfs的操作和Checkpoint进行了解耦，任务运行过程中，每一小批数据就会写入HDFS一次，由于HDFS支持追加写操作，这种形式也不会带来小文件的问题，从而将Checkpoint尽可能的轻量化，避免HDFS抖动和计算量过大带来的Checkpoint超时的问题。
Compaction任务本质上是一个批任务，所以需要和流式写入进行拆分，目前Hudi on Flink支持了异步执行Compaction的操作，我们的线上任务全部使用了这种模式。
在这种模式下，流式任务可以专注于写入，提升吞吐能力和提高写入的稳定性，批式的Compaction任务可以流式任务解耦，弹性伸缩高效的利用计算资源，专注于资源利用率和节约成本。在这一系列的优化过后，我们在一个2百万rps的Kafka数据源上进行了测试，使用了200个并发导入到Hudi。和之前相比，Checkpoint耗时从3-5分钟降低到了1分钟以内，HDFS抖动带来的任务失败率也大幅度下降由于Checkpoint耗时降低，实际用于数据处理的时间变得更多了，数据吞吐量翻了一倍，同时State的存储开销也降到了最低。首先，不同的数据库会将Binlog发送到消息队列中，Flink任务会将所有数据转换成HoodieRecord格式，然后通过哈希索引找到对应的文件ID，通过一层对文件ID的shuffle后，数据到达了写入层，写入算子以追加写的形式将数据频繁的写入到HDFS中，Checkpoint触发后，Flink会将所有的元数据收集到一起，并写入到hudi的元数据系统中，这里就标志了一个Commit提交完成，一个新的Commit会随之开始。用户可以通过Flink Spark Presto等查询引擎，近实时的查询已经提交完成的数据数据湖平台侧托管的Compaction服务会定时提交Flink Batch模式的Compaction任务，对Hudi表进行压缩操作，这个过程对用户无感知并且不影响写入任务。我们这一整套解决方案也会贡献给社区，感兴趣的同学可以关注Hudi社区最新的进展流式数据湖集成框架改造完成后，我们找到了一些典型的落地场景：应用最普遍的就是将线上数据库导入到离线数仓进行分析的场景，和之前的Spark离线链路相比：端到端的数据延迟从一个小时以上降低到了5-10分钟，用户可以进行近实时的数据分析操作。在资源利用率方面，我们模拟了一个Mysql导入离线数仓进行分析的场景，将Flink流式导入Hudi和Spark离线合并的方案进行了对比，在用户小时级查询的场景下，端到端的计算资源大约节约了70%左右。在字节跳动EB级数据量的数仓场景下，这种资源利用率的提升所带来的收益是非常巨大的。
对于基于消息队列和Flink构建实时数仓的用户来说，他们可以把不同数仓层级的实时数据导入到Hudi，这类数据update的情况很多，所以相较于Hive，Hudi可以提供高效率且低成本的Upsert操作，从而用户可以对于全量数据进行近实时查询，避免了一次去重的操作。
这是一个Flink双流Join的场景，很多Flink的用户会使用双流Join来进行实时的字段拼接，在使用这个功能的时候，用户通常会开一个时间窗口，然后将这个时间窗口中来自不同数据源的数据拼接起来，这个字段拼接功能也可以在Hudi的层面实现。我们正在探索一个功能，在Flink中只将不同Topic的数据Union在一起，然后通过Hudi的索引机制，将相同主键的数据都写入到同一个文件当中，然后通过Compaction的操作，将数据进行拼接。这种方式的优点在于，我们可以通过Hudi的索引机制来进行全局字段拼接，不会受到一个窗口的限制。整个拼接逻辑通过HoodiePayload实现，用户可以简单的继承HoodiePayload，然后来开发自己的自定义的拼接逻辑，拼接的时机可以是Compaction任务，也可以是Merge on Read近实时查询，用户可以根据需求场景，灵活的使用计算资源。但是相比Flink双流Join，这种模式会有一个缺点，就是实时性和易用性上要差一些。在这一系列的工作过后，我们对数据湖的未来满怀期待，同时也设立的明确的目标。
首先，我们希望将Hudi作为所有CDC数据源的底层存储，完全替换掉基于Spark的离线合并方案，通过数据集成引擎流式导入，将近实时离线分析的能力带给所有的在线数据库
接着，增量ETL场景也是一个重要的落地场景，对于数据延迟容忍度在分钟级的场景，Hudi可以作为统一存储同时服务于实时链路和离线链路，从而将传统的数仓Lambda架构升级到真正意义上的流批一体。最后，我们希望建设一个智能数据湖平台，这个平台会托管所有数据湖的运维管理，达到自我治理的一个状态，用户则不需要再为运维而烦恼。同时，我们希望提供自动化调优的功能，基于数据的分布找到最佳的配置参数，例如之前提到的不同索引之间的性能取舍问题，我们希望通过算法来找到最佳的配置，从而提高资源利用率，并降低用户的使用门槛。极佳的用户体验也是我们的追求之一，我们希望在平台侧做到一键入湖入仓，大大降低用户的开发成本。
数据湖集成技术也已经通过火山引擎大数据研发治理套件DataLeap对外开放。
产品介绍
火山引擎大数据研发治理套件DataLeap
一站式数据中台套件，帮助用户快速完成数据集成、开发、运维、治理、资产、安全等全套数据中台建设，帮助数据团队有效的降低工作成本和数据维护成本、挖掘数据价值、为企业决策提供数据支撑。后台回复数字“2”了解产品